package com.bigdata.kafka;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;

/**
 * @author xin.ding
 * @date 2021/4/6 10:18
 */
public class KafkaManualAssign {
    private static final Logger logger = LoggerFactory.getLogger(KafkaManualAssign.class);

    private static Properties props = new Properties();
    private static KafkaConsumer<String, String> c1, c2;

    private static final String brokerList = "10.243.141.133:9092";

    static {
        props.put("bootstrap.servers", brokerList);
        props.put("group.id", "test1111");
        props.put("auto.offset.reset", "earliest");
        props.put("enable.auto.commit", "true");
        props.put("session.timeout.ms", "30000");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

        c1 = new KafkaConsumer<>(props);
        c2 = new KafkaConsumer<>(props);
    }

    public static void main(String[] args) {
        TopicPartition tp = new TopicPartition("test1", 0);
        // 采用assign方式显示的为consumer指定需要消费的topic, 具有相同group.id的两个消费者
        // 各自消费了一份数据, 出现了数据的重复消费
        c1.assign(Arrays.asList(tp));
        c2.assign(Arrays.asList(tp));


        // 采用subscribe方式, 利用broker为consumer自动分配topic-partitions,
        // 两个消费者各自消费一个partition, 数据互补, 无交叉.
//         c1.subscribe(Arrays.asList("topic"));
//         c2.subscribe(Arrays.asList("topic"));

        while (true) {
            ConsumerRecords<String, String> msg1 = c1.poll(Duration.ofMillis(1000));
            if (msg1 != null) {
                for (ConsumerRecord m1 : msg1) {
                    logger.info("m1 offset : {} , value : {}", m1.offset(), m1.value());
                }
            }

            logger.info("=====================");
            ConsumerRecords<String, String> msg2 = c2.poll(Duration.ofMillis(1000));
            if (msg2 != null) {
                for (ConsumerRecord m2 : msg2) {
                    logger.info("m2 offset : {} , value : {}", m2.offset(), m2.value());
                }
            }

//            System.exit(0);
        }
    }
}
